package cn.com.greatwall.kafka.service.impl;

import java.io.IOException;
import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import cn.com.greatwall.kafka.domain.Cluster;
import cn.com.greatwall.kafka.repository.ClusterRepository;
import cn.com.greatwall.kafka.service.ClusterService;
import cn.com.greatwall.kafka.service.KafkaService;
import cn.com.greatwall.kafka.service.dto.ConsumerDto;
import cn.com.greatwall.kafka.util.Consumers;
import cn.com.greatwall.kafka.util.KafkaUtil;
import cn.com.greatwall.kafka.util.Producers;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

/**
 * @Author herw
 * @Time 2021-01-13 15:12:18
 * @Version 1.0
 * @Description: TODO(用一句话描述该文件做什么)
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class KafkaServiceImpl implements KafkaService {
    private Map<String, KafkaUtil> cacheMap = new HashMap<>(16);
    private final ClusterRepository clusterRepository;
    @Autowired
    private ClusterService clusterService;

    @Override
    public boolean testConnectionKafka(Cluster resources) {
        boolean flag = false;
        FutureTask<Object> futureTask = new FutureTask<>(new Callable<Object>() {
            @Override
            public Object call() throws Exception {
                boolean address = false;
                KafkaUtil kafka = null;
                try {
                    Properties properties = new Properties();
                    properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, resources.getBroker());
                    kafka = new KafkaUtil(properties);
                    String host = kafka.describeClusterResult().controller().get().host();
                    if (host != null) {
                        address = true;
                    }
                } catch (Exception e) {
                    log.warn("connect kafka error.", e);
                } finally {
                    if (kafka != null) {
                        kafka.close();
                    }
                }
                return address;
            }
        });
        new Thread(futureTask).start();
        Object address = null;
        try {
            address = futureTask.get(3, TimeUnit.SECONDS);
            if (address != null) {
                flag = true;
            }
        } catch (Exception e) {
            log.warn("kafka is inactive.", e);
        }
        return flag;
    }

    @Override
    public KafkaUtil getKafkaUtil(String clusterId) {
        synchronized (cacheMap) {
            Cluster cluster = clusterRepository.findById(Long.parseLong(clusterId)).orElseGet(Cluster::new);
            if (!cacheMap.containsKey(clusterId)) {
                Properties properties = new Properties();
                properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, cluster.getBroker());
                this.cacheMap.put(clusterId.toString(), new KafkaUtil(properties));
            }
            return this.cacheMap.get(clusterId.toString());
        }
    }

    @Override
    public void send(String clusterId, String topicName, String key, String value) throws Exception {
        Cluster cluster = clusterRepository.findById(Long.parseLong(clusterId)).orElseGet(Cluster::new);
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.getBroker());
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        Producers<String, String> producers = new Producers<>(props);
        try {
            producers.send(topicName, key, value);
        } catch (Exception e) {
            log.error("Producer  topic: {} error! key:{}  value:{} ", topicName, key, value, e);
            throw e;
        } finally {
            try {
                producers.close();
            } catch (IOException e) {
                log.error("close Producer error! ");
            }
        }
    }

    @Override
    public ConsumerRecords<String, String> consumer(ConsumerDto resources) {
        Duration timeOut = Duration.ofMillis(Long.parseLong(resources.getWaitTime()));
        boolean isCommit = Boolean.parseBoolean(resources.getIsCommit());
        String clusterId = resources.getClusterId();
        String groupId = resources.getGroupName();
        String topicName = resources.getTopicName();
        int dataSize = Integer.parseInt(resources.getRecordNum());
        boolean isByPartition = Boolean.parseBoolean(resources.getIsByPartition());
        
        if(isByPartition) {
            int partition = Integer.parseInt(resources.getPartition());
            long offset = Long.parseLong(resources.getOffset());
            return consumer(clusterId, groupId, topicName, dataSize, timeOut, isCommit, partition, offset);
        } else {
            return consumer(clusterId, groupId, topicName, dataSize, timeOut, isCommit);
        }
    }
    
    /**
     * 通过指定partition和offset方式消费topic
     * @param clusterId
     * @param groupName
     * @param topicName
     * @param dataSize
     * @param duration
     * @param isCommit
     * @param partition
     * @param offset
     * @return
     */
    private ConsumerRecords<String, String> consumer(String clusterId, String groupName, String topicName, int dataSize, Duration duration, boolean isCommit, Integer partition, Long offset) {
        ConsumerRecords<String, String> result = null;
        
        Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
        String brokers = clusterService.findById(Long.parseLong(clusterId)).getBroker();
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupName);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, dataSize);

        Consumers<String, String> consumer = new Consumers<>(props);
        TopicPartition topicPartition = new TopicPartition(topicName, partition);
        consumer.assign(topicPartition);
        consumer.seek(topicPartition, offset);
        try {
            result = consumer.poll(duration);
            List<ConsumerRecord<String, String>> records = result.records(topicPartition);
            if (!records.isEmpty()) {
                long lastOffset = records.get(records.size() - 1).offset();
                offsets.put(topicPartition, new OffsetAndMetadata(lastOffset + 1));
            }
        } catch (Exception e) {
            log.error("consumer  topic: {} error!", topicName, e);
            throw e;
        } finally {
            try {
                if (isCommit) {
                    consumer.commitByPartition(offsets);
                }
                consumer.close();
            } catch (IOException e) {
                log.error("close consumer error!");
            }
        }
        return result;
    }
    
    /**
     * 普通方式消费topic
     * @param clusterId
     * @param groupName
     * @param topicName
     * @param dataSize
     * @param duration
     * @param isCommit
     * @return
     */
    private ConsumerRecords<String, String> consumer(String clusterId, String groupName, String topicName, int dataSize, Duration duration, boolean isCommit) {
        ConsumerRecords<String, String> result = null;
        
        String brokers = clusterService.findById(Long.parseLong(clusterId)).getBroker();
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupName);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, dataSize);
        Consumers<String, String> consumer = new Consumers<>(props);
        consumer.subscribe(topicName);
        try {
            result = consumer.poll(duration);
        } catch (Exception e) {
            log.error("consumer  topic: {} error!", topicName, e);
            throw e;
        } finally {
            try {
                if (isCommit) {
                    consumer.commit();
                }
                consumer.close();
            } catch (IOException e) {
                log.error("close consumer error!");
            }
        }
        
        return result;
    }
}
